/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.trans.worker;

import cn.ac.iie.ulss.data.trans.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.trans.util.Meta;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.log4j.Logger;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;

/**
 *
 * @author Li Mingyang
 */
public class DataProducer extends Thread {

    static Logger logger = Logger.getLogger(DataProducer.class.getName());

    private Meta meta;
    private DefaultMQProducer producer;

    public DataProducer(Meta meta, DefaultMQProducer producer) {
        this.meta = meta;
        this.producer = producer;
    }

    @Override
    public void run() {
        Protocol protocol = null;
        try {
            protocol = Protocol.parse(new File("docs.json"));
        } catch (IOException ex) {
            logger.error(ex.getMessage(), ex);
        }
        Schema docsSchema = protocol.getType("docs");
        DatumReader<GenericRecord> docsReader = new GenericDatumReader(docsSchema);
        while (true) {
            byte[] data = (byte[]) RuntimeEnv.TOPIC_TO_DATAPOOL.get(meta.getDataName()).poll();
            if (data == null) {
                logger.info("the LinkedBlockingQueue for " + meta.getDataName() + " is empty and wait.");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            } else {
                ByteArrayInputStream docsbis = new ByteArrayInputStream(data);
                BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                GenericRecord docsRecord = new GenericData.Record(docsSchema);
                try {
                    docsReader.read(docsRecord, docsbd);
                } catch (IOException ex) {
                    logger.error(ex.getMessage(), ex);
                }
                GenericArray docSet = (GenericArray) docsRecord.get("doc_set");
                this.sendMessage(meta.getDestinationParms().split("[-]")[1], data, docSet.size());
            }
        }
    }

    public void sendMessage(String topic, byte[] pData, long count) {
        long bg = System.currentTimeMillis();
        SendResult sendResult = null;
        Message msg = new Message(topic, pData);
        while ((sendResult == null) || (sendResult.getSendStatus() != SendStatus.SEND_OK)) {
            try {
                sendResult = producer.send(msg);
                if ((sendResult == null) || (sendResult.getSendStatus() == SendStatus.FLUSH_DISK_TIMEOUT)) {
                    if (sendResult == null) {
                        logger.warn("send message fail,the send result is null,will sleep and retry ...");
                    } else {
                        logger.warn("send message fail one time for write disk fail,will sleep and retry,the information is " + producer.getClientIP() + " " + producer.getProducerGroup());
                    }
                    try {
                        Thread.sleep(200L);
                    } catch (Exception e) {
                    }
                    continue;
                }
                logger.info("send " + count + " records and size " + pData.length + " bytes,use " + (System.currentTimeMillis() - bg) + " ms for " + topic + ":" + sendResult);
            } catch (Exception ex) {
                logger.error(ex + ",the information is:topic --> " + topic + ",body size is " + pData.length + "，will retry,", ex);
                try {
                    Thread.sleep(200L);
                } catch (Exception e) {
                }
                continue;
            } finally {
            }
        }
    }
}
